package com.rn.gypsophila.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.stereotype.Component;

/**
 * Description: 消费者
 * https://blog.csdn.net/yuanlong122716/article/details/105160545/
 *
 * @author 然诺
 * @date 2021/01/29
 */
@Component
public class KafkaConsumer {

	@Bean
	public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
		return (message, exception, consumer) -> {
			System.out.println("消费异常：" + message.getPayload());
			return null;
		};
	}

	/**
	 * 简单消费消息
	 */
	@KafkaListener(topics = {"topic_1"})
	public void receiveMessage(String message) {
		System.out.println("收到消息：" + message);
	}

	/**
	 * 1.消息过滤器
	 * 2.监听异常处理器
	 */
	@KafkaListener(topics = {"topic_2"}, containerFactory = "filterContainerFactory", errorHandler = "consumerAwareErrorHandler")
	public void onMessage(ConsumerRecord<?, ?> record) throws Exception {
		System.out.println("收到消息：" + record.topic() + "-" + record.partition() + "-" + record.value());
		throw new Exception("简单消费-模拟异常");
	}

	/**
	 * 消费消息
	 */
	@KafkaListener(topics = {"topic_3"})
	public void receiveMessage(ConsumerRecord<?, ?> record) {
		System.out.println("收到消息：" + record.topic() + "-" + record.partition() + "-" + record.value());
	}

}
